package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.function.DimAsyncFunction;
import com.atguigu.gmall.realtime.bean.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.MyClickhouseUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import com.atguigu.gmall.realtime.util.TimestampLtz3CompareUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @author Felix
 * @date 2022/11/26
 * 交易域：商品SKU粒度下单业务过程聚合统计
 * 维度：SKU
 * 度量：下单独立用户、订单数、原始金额、活动减免金额、优惠券减免金额、实付金额
 * 需要启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、redis、clickhouse
 *      DwdTradeOrderPreProcess、DwdTradeOrderDetail、DwsTradeSkuOrderWindow
 * 开发流程
 *      基本环境准备
 *      检查点设置
 *      从kafka的下单主题中读取数据
 *      对读取的数据进行类型转换    jsonStr->jsonObj
 *      去重
 *          为什么会产生重复数据？
 *              因为我们是从下单主题中读取数据的，下单主题的数据来源于订单预处理表，订单预处理表的数据是订单、订单明细、订单明细活动、
 *              订单明细优惠券、字典表5张表进行关联得到的。在和明细活动以及明细优惠券进行连接时候，使用的是左外连接，如果左外连接的时候
 *              左表数据先到，右表数据后到，会产生如下数据
 *                      +I  左表      null
 *                      -D  左表      null
 *                      +I  左表      右表,
 *              数据发送到kafka主题后，kafka接收到的数据为
 *                      左表      null
 *                      null
 *                      左表      右表
 *              对于null消息，如果使用kafka连接器读取数据的话，可以直接将null过滤掉，但是存在数据的重复
 *          去重思路
 *              按照唯一键（订单明细id）进行分组
 *              使用Flink的状态编程 + 定时器完成去重操作
 *              当第一条数据来的时候，将数据放到状态中，并注册一个5秒之后执行的定时器；后面的数据来的时候，用状态的数据和当前数据的时间进行比较，
 *              将时间大的数据放到状态中。等定时器被触发的时候，将状态中的数据下下游传递
 *      类型转换    jsonObj->实体类对象
 *      判断是否下单独立用户
 *          按照用户id进行分组
 *          使用Flink的状态编程进行判断
 *      指定Watermark以及提取事件时间字段
 *      按照要统计的维度sku进行分组
 *      开窗
 *      聚合计算
 *      ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 *      和维度进行关联
 *          基本维度关联
 *              PhoenixUtil
 *                  List<T> queryList(String sql,Class<T> clz)
 *              DimUtil
 *                  JSONObject getDimInfoNoCache(String sql,Tuple2... params)
 *          优化1：旁路缓存
 *              思路：先从缓存中查询数据，如果缓存中存在要关联的维度，直接将维度作为返回值返回(缓存命中)；
 *                  如果缓存中不存在要关联的维度，再发送请求到Phoenix表中进行查询，并将查询出来的维度放到缓存中缓存起来，方便下次查询
 *              选型：
 *                  redis   性能好，维护性好 √
 *                  状态  性能很好，维护性差
 *              DimUtil
 *                      getDimInfo
 *          优化2：异步IO
 *              要想提升对流中数据的处理能力，可以加大并行度，但是加大并行度意味需要更多的硬件资源，不可能无限制的加大。
 *              在单个并行度上，默认对流中数据进行处理的时候，使用的是同步的方式，处理完一个元素之后，再处理下一个元素
 *              可以使用Flink提供异步处理API，同时对流中的元素进行处理
 *              AsyncDataStream.[un]orderedWait(
 *                  流，
 *                  异步操作,
 *                  超时时间,
 *                  时间单位
 *              )
 *              class DimAsyncFunction extends RichAsyncFunction{
 *                  asyncInvoke{
 *                      开启多个线程，发送异步请求
 *                      模板方法设计模式：在父类中定义完成某一个功能的核心算法骨架，将具体的实现延迟到子类中去完成；
 *                      在不改变父类核心算法骨架的前提下，每一个子类都可以有自己不同的实现。
 *                  }
 *              }
 *      将结果写到Clickhouse中
 */
public class DwsTradeSkuOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //TODO 2.检查点相关的设置(略)
        //TODO 3.从kafka下单主题中读取数据
        //3.1 指定消费的主题以及消费者组
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_sku_order_group";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);

        //TODO 4.对读取的数据进行类型转换 jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        // {"create_time":"2022-11-20 09:12:34","sku_num":"1","activity_rule_id":"4","split_original_amount":"9197.0000",
        // "sku_id":"12","date_id":"2022-11-20","source_type_name":"用户查询","user_id":"65","province_id":"33",
        // "source_type_code":"2401","row_op_ts":"2022-11-28 01:12:34.809Z","activity_id":"2","sku_name":"Apple iPhone ",
        // "id":"220","order_id":"96","split_activity_amount":"179.69","split_total_amount":"9017.31","ts":"1669597954"}
        // jsonObjDS.print(">>>");

        //TODO 5.按照唯一键(订单明细id)进行分组
        KeyedStream<JSONObject, String> orderDetailIdKeyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));
        //TODO 6.使用Flink的状态 + 定时器 完成去重
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
            new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                private ValueState<JSONObject> lastValueState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    lastValueState
                        = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("lastValueState", JSONObject.class));
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                    //获取状态中是否有数据进来
                    JSONObject lastValue = lastValueState.value();
                    if (lastValue == null) {
                        //如果状态中没有数据，说明当前数据是第一条数据，注册一个定时器，并将当前数据放到状态中
                        long currentProcessingTime = ctx.timerService().currentProcessingTime();
                        ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                        lastValueState.update(jsonObj);
                    } else {
                        //如果状态中已经有数据进来，说明产生重复了，对比状态中数据的时间和当前数据的时间 "row_op_ts":"2022-11-28 01:12:34.809Z"
                        String lastRowOpTs = lastValue.getString("row_op_ts");
                        String curRowOpTs = jsonObj.getString("row_op_ts");
                        if (TimestampLtz3CompareUtil.compare(lastRowOpTs, curRowOpTs) <= 0) {
                            lastValueState.update(jsonObj);
                        }
                    }
                }

                @Override
                public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                    //当定时器被触发后，将状态中数据写到下游
                    JSONObject value = lastValueState.value();
                    if (value != null) {
                        out.collect(value);
                    }
                    lastValueState.clear();
                }
            }
        );

        //TODO 7.转换流中结构 jsonObj->实体类对象
        SingleOutputStreamOperator<TradeSkuOrderBean> orderBeanDS = distinctDS.map(
            new MapFunction<JSONObject, TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean map(JSONObject jsonObj) throws Exception {
                    // {"create_time":"2022-11-20 09:12:34","sku_num":"1","activity_rule_id":"4","split_original_amount":"9197.0000",
                    // "sku_id":"12","date_id":"2022-11-20","source_type_name":"用户查询","user_id":"65","province_id":"33",
                    // "source_type_code":"2401","row_op_ts":"2022-11-28 01:12:34.809Z","activity_id":"2","sku_name":"Apple iPhone ",
                    // "id":"220","order_id":"96","split_activity_amount":"179.69","split_total_amount":"9017.31","ts":"1669597954"}
                    String skuId = jsonObj.getString("sku_id");
                    String orderId = jsonObj.getString("order_id");
                    String userId = jsonObj.getString("user_id");
                    Double originalAmount = jsonObj.getDouble("split_original_amount");
                    Double activityAmount = jsonObj.getDouble("split_activity_amount");
                    Double couponAmount = jsonObj.getDouble("split_coupon_amount");
                    Double totalAmount = jsonObj.getDouble("split_total_amount");
                    Long ts = jsonObj.getLong("ts") * 1000;

                    TradeSkuOrderBean orderBean = TradeSkuOrderBean.builder()
                        .skuId(skuId)
                        .orderIdSet(new HashSet<>(Collections.singleton(orderId)))
                        .userId(userId)
                        .orderUuCount(0L)
                        .originalAmount(originalAmount)
                        .activityAmount(activityAmount == null ? 0.0 : activityAmount)
                        .couponAmount(couponAmount == null ? 0.0 : couponAmount)
                        .orderAmount(totalAmount)
                        .ts(ts)
                        .build();
                    return orderBean;
                }
            }
        );
        //TODO 8.按照userId进行分组
        KeyedStream<TradeSkuOrderBean, String> userIdKeyedDS = orderBeanDS.keyBy(TradeSkuOrderBean::getUserId);
        //TODO 9.统计独立用户数
        SingleOutputStreamOperator<TradeSkuOrderBean> uuCtDS = userIdKeyedDS.process(
            new KeyedProcessFunction<String, TradeSkuOrderBean, TradeSkuOrderBean>() {
                private ValueState<String> lastOrderDateState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<String>("lastOrderDateState", String.class);
                    valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                    lastOrderDateState = getRuntimeContext().getState(valueStateDescriptor);
                }

                @Override
                public void processElement(TradeSkuOrderBean orderBean, Context ctx, Collector<TradeSkuOrderBean> out) throws Exception {
                    //获取状态中的上次下单日期
                    String lastOrderDate = lastOrderDateState.value();
                    //获取当前这次下单日期
                    Long ts = orderBean.getTs();
                    String curOrderDate = DateFormatUtil.toDate(ts);
                    if (StringUtils.isEmpty(lastOrderDate) || !curOrderDate.equals(lastOrderDate)) {
                        orderBean.setOrderUuCount(1L);
                        lastOrderDateState.update(curOrderDate);
                    }
                    out.collect(orderBean);
                }
            }
        );
        //TODO 10.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<TradeSkuOrderBean> withWatermarkDS = uuCtDS.assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<TradeSkuOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<TradeSkuOrderBean>() {
                        @Override
                        public long extractTimestamp(TradeSkuOrderBean orderBean, long recordTimestamp) {
                            return orderBean.getTs();
                        }
                    }
                )
        );
        //TODO 11.按照skuId分组
        KeyedStream<TradeSkuOrderBean, String> skuIdKeyedDS = withWatermarkDS.keyBy(TradeSkuOrderBean::getSkuId);

        //TODO 12.开窗
        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDS = skuIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        //TODO 13.聚合
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDS = windowDS.reduce(
            new ReduceFunction<TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                    value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                    value1.setOrderUuCount(value1.getOrderUuCount() + value2.getOrderUuCount());
                    value1.setOriginalAmount(value1.getOriginalAmount() + value2.getOriginalAmount());
                    value1.setActivityAmount(value1.getActivityAmount() + value2.getActivityAmount());
                    value1.setCouponAmount(value1.getCouponAmount() + value2.getCouponAmount());
                    value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());
                    return value1;
                }
            },
            new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                @Override
                public void apply(String groupId, TimeWindow window, Iterable<TradeSkuOrderBean> input, Collector<TradeSkuOrderBean> out) throws Exception {
                    for (TradeSkuOrderBean orderBean : input) {
                        orderBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        orderBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        orderBean.setOrderCount((long) orderBean.getOrderIdSet().size());
                        orderBean.setTs(System.currentTimeMillis());
                        out.collect(orderBean);
                    }
                }
            }
        );
        //TradeSkuOrderBean(stt=2022-11-28 11:48:10, edt=2022-11-28 11:48:20, trademarkId=null, trademarkName=null, category1Id=null,
        // category1Name=null, category2Id=null, category2Name=null, category3Id=null, category3Name=null, orderIdSet=[111],
        // userId=26, skuId=12, skuName=null, spuId=null, spuName=null, orderUuCount=0, orderCount=1, originalAmount=18394.0,
        // activityAmount=1200.0, couponAmount=0.0, orderAmount=17194.0, ts=1669607592298)
        // reduceDS.print(">>>>");
        //TODO 14.关联Sku维度
       /* SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
            new MapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                    //根据流中的对象获取要关联的维度的主键
                    String skuId = orderBean.getSkuId();
                    //根据维度的主键到维度表中获取维度数据
                    // ID,SPU_ID,PRICE,SKU_NAME,SKU_DESC,WEIGHT,TM_ID,CATEGORY3_ID,SKU_DEFAULT_IMG,IS_SALE,CREATE_TIME
                    JSONObject dimInfoJsonObj = DimUtil.getDimInfo("DIM_SKU_INFO", skuId);
                    //将获取到的维度对象的属性补充到流中的对象上
                    orderBean.setSkuName(dimInfoJsonObj.getString("SKU_NAME"));
                    orderBean.setSpuId(dimInfoJsonObj.getString("SPU_ID"));
                    orderBean.setCategory3Id(dimInfoJsonObj.getString("CATEGORY3_ID"));
                    orderBean.setTrademarkId(dimInfoJsonObj.getString("TM_ID"));
                    return orderBean;
                }
            }
        );
        withSkuInfoDS.print(">>");*/
        //将异步 I/O 操作应用于 DataStream 作为 DataStream 的一次转换操作。
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
            reduceDS,
            //实现分发请求的 AsyncFunction
            new DimAsyncFunction<TradeSkuOrderBean>("dim_sku_info") {
                @Override
                public void join(JSONObject dimInfoJsonObj, TradeSkuOrderBean orderBean) {
                    orderBean.setSkuName(dimInfoJsonObj.getString("SKU_NAME"));
                    orderBean.setSpuId(dimInfoJsonObj.getString("SPU_ID"));
                    orderBean.setCategory3Id(dimInfoJsonObj.getString("CATEGORY3_ID"));
                    orderBean.setTrademarkId(dimInfoJsonObj.getString("TM_ID"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSkuId();
                }
            },
            60,
            TimeUnit.SECONDS
        );
        //TODO 15.关联Spu维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withSpuInfoDS = AsyncDataStream.unorderedWait(
            withSkuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_spu_info") {
                @Override
                public void join(JSONObject dimInfoJsonObj, TradeSkuOrderBean orderBean) {
                    orderBean.setSpuName(dimInfoJsonObj.getString("SPU_NAME"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSpuId();
                }
            },
            60, TimeUnit.SECONDS
        );
        //TODO 16.关联tm维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withTmDS = AsyncDataStream.unorderedWait(
            withSpuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_trademark") {
                @Override
                public void join(JSONObject dimInfoJsonObj, TradeSkuOrderBean orderBean) {
                    orderBean.setTrademarkName(dimInfoJsonObj.getString("TM_NAME"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getTrademarkId();
                }
            },
            60, TimeUnit.SECONDS
        );
        //TODO 17.关联category3维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withCategory3Stream = AsyncDataStream.unorderedWait(
            withTmDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category3".toUpperCase()) {
                @Override
                public void join(JSONObject jsonObj,TradeSkuOrderBean javaBean) {
                    javaBean.setCategory3Name(jsonObj.getString("name".toUpperCase()));
                    javaBean.setCategory2Id(jsonObj.getString("category2_id".toUpperCase()));
                }

                @Override
                public String getKey(TradeSkuOrderBean javaBean) {
                    return javaBean.getCategory3Id();
                }
            },
            5 * 60, TimeUnit.SECONDS
        );

        //TODO 18.关联category2维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withCategory2Stream = AsyncDataStream.unorderedWait(
            withCategory3Stream,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category2".toUpperCase()) {
                @Override
                public void join(JSONObject jsonObj,TradeSkuOrderBean javaBean)  {
                    javaBean.setCategory2Name(jsonObj.getString("name".toUpperCase()));
                    javaBean.setCategory1Id(jsonObj.getString("category1_id".toUpperCase()));
                }

                @Override
                public String getKey(TradeSkuOrderBean javaBean) {
                    return javaBean.getCategory2Id();
                }
            },
            5 * 60, TimeUnit.SECONDS
        );

        //TODO 19.关联category1维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withCategory1Stream = AsyncDataStream.unorderedWait(
            withCategory2Stream,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category1".toUpperCase()) {
                @Override
                public void join( JSONObject jsonObj,TradeSkuOrderBean javaBean) {
                    javaBean.setCategory1Name(jsonObj.getString("name".toUpperCase()));
                }

                @Override
                public String getKey(TradeSkuOrderBean javaBean) {
                    return javaBean.getCategory1Id();
                }
            },
            5 * 60, TimeUnit.SECONDS
        );

        //TODO 20.将聚合的结果写到CK中
        withCategory1Stream.print(">>>");
        withCategory1Stream.addSink(MyClickhouseUtil.getSinkFunction("insert into dws_trade_sku_order_window values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));
        env.execute();
    }
}
